/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.broker.transaction.queue;

import io.opentelemetry.api.common.Attributes;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;

import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CONSUMER_GROUP;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_SYSTEM;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_TOPIC;

public class TransactionalMessageBridge {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME);

    /**
     * key是HALF消息queueId，value是OP消息mq
     */
    private final ConcurrentHashMap<Integer, MessageQueue> opQueueMap = new ConcurrentHashMap<>();
    private final BrokerController brokerController;
    private final MessageStore store;
    private final SocketAddress storeHost;

    public TransactionalMessageBridge(BrokerController brokerController, MessageStore store) {
        try {
            this.brokerController = brokerController;
            this.store = store;
            this.storeHost =
                new InetSocketAddress(brokerController.getBrokerConfig().getBrokerIP1(),
                    brokerController.getNettyServerConfig().getListenPort());
        } catch (Exception e) {
            LOGGER.error("Init TransactionBridge error", e);
            throw new RuntimeException(e);
        }

    }

    /**
     * 根据消息队列实例获取消费偏移量
     * @param mq
     * @return
     */
    public long fetchConsumeOffset(MessageQueue mq) {
        // 通过消费偏移量查询消费偏移量
        // 入参：
        //     - 消费组名称："CID_RMQ_SYS_TRANS"
        //     - 消息队列的主题名称
        //     - 消息队列ID
        long offset = brokerController.getConsumerOffsetManager().queryOffset(TransactionalMessageUtil.buildConsumerGroup(),
            mq.getTopic(), mq.getQueueId());
        if (offset == -1) {
            // 如果没有找到消息偏移量，则返回该主题该消息队列的最小偏移量
            offset = store.getMinOffsetInQueue(mq.getTopic(), mq.getQueueId());
        }
        // 返回消费偏移量
        return offset;
    }

    /**
     * 从指定主题名称获取MQ集合
     * @param topic
     * @return
     */
    public Set<MessageQueue> fetchMessageQueues(String topic) {
        Set<MessageQueue> mqSet = new HashSet<>();
        // 获取主题配置信息
        TopicConfig topicConfig = selectTopicConfig(topic);
        // 如果找到了主题配置信息，并且主题的读消息队列个数大于0
        if (topicConfig != null && topicConfig.getReadQueueNums() > 0) {
            // 则遍历读MQ
            for (int i = 0; i < topicConfig.getReadQueueNums(); i++) {
                // 创建MQ实例
                MessageQueue mq = new MessageQueue();
                // 设置主题名称
                mq.setTopic(topic);
                // 设置brokername
                mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName());
                // 设置mqId
                mq.setQueueId(i);
                // 将mq添加到Set集合中
                mqSet.add(mq);
            }
        }
        // 返回找到的MQ集合
        return mqSet;
    }

    public void updateConsumeOffset(MessageQueue mq, long offset) {
        this.brokerController.getConsumerOffsetManager().commitOffset(
            RemotingHelper.parseSocketAddressAddr(this.storeHost), TransactionalMessageUtil.buildConsumerGroup(), mq.getTopic(),
            mq.getQueueId(), offset);
    }

    /**
     *
     * @param queueId
     * @param offset
     * @param nums 获取多少条消息
     * @return
     */
    public PullResult getHalfMessage(int queueId, long offset, int nums) {
        String group = TransactionalMessageUtil.buildConsumerGroup();
        String topic = TransactionalMessageUtil.buildHalfTopic();
        SubscriptionData sub = new SubscriptionData(topic, "*");
        return getMessage(group, topic, queueId, offset, nums, sub);
    }

    /**
     * 读取OP消息
     * @param queueId 消息队列ID
     * @param offset 偏移量
     * @param nums 读取消息的个数
     * @return 读取到的消息的封装对象
     */
    public PullResult getOpMessage(int queueId, long offset, int nums) {
        // 消费组名称："CID_RMQ_SYS_TRANS"
        String group = TransactionalMessageUtil.buildConsumerGroup();
        // 主题名称："RMQ_SYS_TRANS_OP_HALF_TOPIC"
        String topic = TransactionalMessageUtil.buildOpTopic();
        // 订阅数据
        SubscriptionData sub = new SubscriptionData(topic, "*");
        return getMessage(group, topic, queueId, offset, nums, sub);
    }

    /**
     *
     * @param group
     * @param topic
     * @param queueId
     * @param offset
     * @param nums
     * @param sub
     * @return
     */
    private PullResult getMessage(String group, String topic, int queueId, long offset, int nums,
        SubscriptionData sub) {
        // 从消息存储获取消息
        GetMessageResult getMessageResult = store.getMessage(group, topic, queueId, offset, nums, null);

        if (getMessageResult != null) {
            PullStatus pullStatus = PullStatus.NO_NEW_MSG;
            List<MessageExt> foundList = null;
            switch (getMessageResult.getStatus()) {
                case FOUND:
                    pullStatus = PullStatus.FOUND;
                    foundList = decodeMsgList(getMessageResult);
                    this.brokerController.getBrokerStatsManager().incGroupGetNums(group, topic,
                        getMessageResult.getMessageCount());
                    this.brokerController.getBrokerStatsManager().incGroupGetSize(group, topic,
                        getMessageResult.getBufferTotalSize());
                    this.brokerController.getBrokerStatsManager().incBrokerGetNums(topic, getMessageResult.getMessageCount());
                    if (foundList == null || foundList.size() == 0) {
                        break;
                    }
                    this.brokerController.getBrokerStatsManager().recordDiskFallBehindTime(group, topic, queueId,
                        this.brokerController.getMessageStore().now() - foundList.get(foundList.size() - 1)
                            .getStoreTimestamp());

                    Attributes attributes = BrokerMetricsManager.newAttributesBuilder()
                        .put(LABEL_TOPIC, topic)
                        .put(LABEL_CONSUMER_GROUP, group)
                        .put(LABEL_IS_SYSTEM, TopicValidator.isSystemTopic(topic) || MixAll.isSysConsumerGroup(group))
                        .build();
                    BrokerMetricsManager.messagesOutTotal.add(getMessageResult.getMessageCount(), attributes);
                    BrokerMetricsManager.throughputOutTotal.add(getMessageResult.getBufferTotalSize(), attributes);

                    break;
                case NO_MATCHED_MESSAGE:
                    pullStatus = PullStatus.NO_MATCHED_MSG;
                    LOGGER.warn("No matched message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
                        getMessageResult.getStatus(), topic, group, offset);
                    break;
                case NO_MESSAGE_IN_QUEUE:
                case OFFSET_OVERFLOW_ONE:
                    pullStatus = PullStatus.NO_NEW_MSG;
                    LOGGER.warn("No new message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
                        getMessageResult.getStatus(), topic, group, offset);
                    break;
                case MESSAGE_WAS_REMOVING:
                case NO_MATCHED_LOGIC_QUEUE:
                case OFFSET_FOUND_NULL:
                case OFFSET_OVERFLOW_BADLY:
                case OFFSET_TOO_SMALL:
                    pullStatus = PullStatus.OFFSET_ILLEGAL;
                    LOGGER.warn("Offset illegal. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
                        getMessageResult.getStatus(), topic, group, offset);
                    break;
                default:
                    assert false;
                    break;
            }

            return new PullResult(pullStatus, getMessageResult.getNextBeginOffset(), getMessageResult.getMinOffset(),
                getMessageResult.getMaxOffset(), foundList);

        } else {
            LOGGER.error("Get message from store return null. topic={}, groupId={}, requestOffset={}", topic, group,
                offset);
            return null;
        }
    }

    private List<MessageExt> decodeMsgList(GetMessageResult getMessageResult) {
        List<MessageExt> foundList = new ArrayList<>();
        try {
            List<ByteBuffer> messageBufferList = getMessageResult.getMessageBufferList();
            for (ByteBuffer bb : messageBufferList) {
                MessageExt msgExt = MessageDecoder.decode(bb, true, false);
                if (msgExt != null) {
                    foundList.add(msgExt);
                }
            }

        } finally {
            getMessageResult.release();
        }

        return foundList;
    }

    public PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner) {
        return store.putMessage(parseHalfMessageInner(messageInner));
    }

    public CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner) {
        return store.asyncPutMessage(parseHalfMessageInner(messageInner));
    }

    /**
     * 解析HALF内部消息
     * @param msgInner HALF消息
     * @return 返回HALF消息的内部表示
     */
    private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
        // 获取HALF消息唯一ID
        String uniqId = msgInner.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
        if (uniqId != null && !uniqId.isEmpty()) {
            // 如果存在唯一ID，则设置消息的 "__transactionId__" 为唯一ID
            MessageAccessor.putProperty(msgInner, TransactionalMessageUtil.TRANSACTION_ID, uniqId);
        }
        // 设置消息的 "REAL_TOPIC" 的值为 消息的目标主题名（将目标主题名称折叠为消息属性）
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
        // 设置消息的 "REAL_QID" 的值为 消息的目标消息队列ID（将目标队列ID折叠为消息属性）
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
            String.valueOf(msgInner.getQueueId()));
        // 将消息的系统标记设置为非事务消息
        msgInner.setSysFlag(
            MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
        // 设置消息的主题："RMQ_SYS_TRANS_HALF_TOPIC"
        msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
        // 将队列ID设置为0
        msgInner.setQueueId(0);
        // 将消息的属性集合设置为消息的propertiesString属性值
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
        // 返回修改后的消息
        return msgInner;
    }

    /**
     * 存储OP消息的内部表示并返回结果
     * @param messageInner OP消息的内部表示
     * @return 返回结果
     */
    public PutMessageResult putMessageReturnResult(MessageExtBrokerInner messageInner) {
        LOGGER.debug("[BUG-TO-FIX] Thread:{} msgID:{}", Thread.currentThread().getName(), messageInner.getMsgId());
        // 存储OP消息，获取结果
        PutMessageResult result = store.putMessage(messageInner);

        // 如果保存成功
        if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
            this.brokerController.getBrokerStatsManager().incTopicPutNums(messageInner.getTopic());
            this.brokerController.getBrokerStatsManager().incTopicPutSize(messageInner.getTopic(),
                    result.getAppendMessageResult().getWroteBytes());
            this.brokerController.getBrokerStatsManager().incBrokerPutNums();
        }
        // 返回result
        return result;
    }

    /**
     * 存储消息
     * @param messageInner 内部消息
     * @return true表示成功，false表示失败
     */
    public boolean putMessage(MessageExtBrokerInner messageInner) {
        // 存储消息，获取结果
        PutMessageResult putMessageResult = store.putMessage(messageInner);
        // 如果PUT_OK，则返回true
        if (putMessageResult != null
            && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
            return true;
        } else {
            LOGGER.error("Put message failed, topic: {}, queueId: {}, msgId: {}",
                messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId());
            // 否则false
            return false;
        }
    }

    public MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt) {
        MessageExtBrokerInner msgInner = renewHalfMessageInner(msgExt);
        String queueOffsetFromPrepare = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET);
        if (null != queueOffsetFromPrepare) {
            MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET,
                    queueOffsetFromPrepare);
        } else {
            MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET,
                String.valueOf(msgExt.getQueueOffset()));
        }

        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));

        return msgInner;
    }

    /**
     * 将MessageExt转换为MessageExtBrokerInner消息
     * @param msgExt
     * @return
     */
    public MessageExtBrokerInner renewHalfMessageInner(MessageExt msgExt) {
        // 创建消息
        MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
        msgInner.setTopic(msgExt.getTopic());
        msgInner.setBody(msgExt.getBody());
        msgInner.setQueueId(msgExt.getQueueId());
        msgInner.setMsgId(msgExt.getMsgId());
        msgInner.setSysFlag(msgExt.getSysFlag());
        msgInner.setTags(msgExt.getTags());
        msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(msgInner.getTags()));
        MessageAccessor.setProperties(msgInner, msgExt.getProperties());
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
        msgInner.setBornTimestamp(msgExt.getBornTimestamp());
        msgInner.setBornHost(msgExt.getBornHost());
        msgInner.setStoreHost(msgExt.getStoreHost());
        msgInner.setWaitStoreMsgOK(false);
        return msgInner;
    }

    /**
     * 将OP消息转换为OP消息的内部表示：
     * <ol>
     * <li>创建OP消息的内部表示</li>
     * <li>设置主题名称为OP消息的主题名称</li>
     * <li>设置消息体</li>
     * <li>设置MQ ID为指定MQ的ID</li>
     * <li>设置标签为OP消息的标签</li>
     * <li>设置标签码</li>
     * <li>设置系统标记</li>
     * <li>设置所有属性</li>
     * <li>设置所有属性字符串</li>
     * <li>设置创建时间戳：当前时间</li>
     * <li>设置创建主机名称</li>
     * <li>设置存储主机名称为当前主机名称</li>
     * <li>设置不需要等待存储OK</li>
     * <li>设置唯一ID</li>
     * <li>返回OP消息的内部表示</li>
     * </ol>
     * @param message OP消息
     * @param messageQueue OP QUEUE
     * @return 返回OP消息的内部表示
     */
    private MessageExtBrokerInner makeOpMessageInner(Message message, MessageQueue messageQueue) {
        // 创建OP消息
        MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
        // 设置主题名称为OP主题名称：
        msgInner.setTopic(message.getTopic());
        // 设置消息体
        msgInner.setBody(message.getBody());
        // 设置MQ ID
        msgInner.setQueueId(messageQueue.getQueueId());
        // 设置标签
        msgInner.setTags(message.getTags());
        // 设置标签的标签码
        msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(msgInner.getTags()));
        // 设置系统标记
        msgInner.setSysFlag(0);
        // 设置所有属性
        MessageAccessor.setProperties(msgInner, message.getProperties());
        // 设置所有属性字符串
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(message.getProperties()));
        // 设置创建时间戳：当前时间
        msgInner.setBornTimestamp(System.currentTimeMillis());
        // 设置创建主机名称
        msgInner.setBornHost(this.storeHost);
        // 设置存储消息的主机名称
        msgInner.setStoreHost(this.storeHost);
        // 设置不需要等待存储OK
        msgInner.setWaitStoreMsgOK(false);
        // 设置唯一ID
        MessageClientIDSetter.setUniqID(msgInner);
        // 返回OP消息的内部表示
        return msgInner;
    }

    private TopicConfig selectTopicConfig(String topic) {
        TopicConfig topicConfig = brokerController.getTopicConfigManager().selectTopicConfig(topic);
        if (topicConfig == null) {
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
                topic, 1, PermName.PERM_WRITE | PermName.PERM_READ, 0);
        }
        return topicConfig;
    }

    /**
     * 写OP消息
     * @param queueId HALF消息的queueId
     * @param message OP消息
     * @return
     */
    public boolean writeOp(Integer queueId,Message message) {
        // 从op队列map中获取queueId对应的OP MQ
        MessageQueue opQueue = opQueueMap.get(queueId);
        if (opQueue == null) {
            // 如果没有找到，就根据HALF消息创建OP MQ
            /*
              根据HALF信息获取OpQueue：
                创建opQueue对象
                设置主题名称："RMQ_SYS_TRANS_OP_HALF_TOPIC"
                设置broker名称：brokerName
                设置queueId：queueId
                返回opQueue
             */
            opQueue = getOpQueueByHalf(queueId, this.brokerController.getBrokerConfig().getBrokerName());
            // 放到OP MQ map集合中，如果有返回值，表示有其他线程放进去了
            MessageQueue oldQueue = opQueueMap.putIfAbsent(queueId, opQueue);
            if (oldQueue != null) {
                // 如果返回值不是null，表示当前添加失败，返回的是现有的，因此赋值
                opQueue = oldQueue;
            }
        }

        // 将OP消息转换为OP消息的内部表示，并保存消息
        PutMessageResult result = putMessageReturnResult(makeOpMessageInner(message, opQueue));
        // 如果OK，则返回true
        if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
            return true;
        }
        // 否则返回false。
        return false;
    }

    /**
     * 根据HALF信息获取OpQueue：
     * <ol>
     * <li>创建opQueue对象</li>
     * <li>设置主题名称："RMQ_SYS_TRANS_OP_HALF_TOPIC"</li>
     * <li>设置broker名称：brokerName</li>
     * <li>设置queueId：queueId</li>
     * <li>返回opQueue</li>
     * </ol>
     * @param queueId HALF消息的queueId
     * @param brokerName HALF消息所在的brokerName
     * @return
     */
    private MessageQueue getOpQueueByHalf(Integer queueId, String brokerName) {
        MessageQueue opQueue = new MessageQueue();
        // "RMQ_SYS_TRANS_OP_HALF_TOPIC"
        opQueue.setTopic(TransactionalMessageUtil.buildOpTopic());
        opQueue.setBrokerName(brokerName);
        opQueue.setQueueId(queueId);
        return opQueue;
    }

    /**
     * 根据偏移量查找消息：
     * <ol>
     * <li>首先通过偏移量，从内存映射文件中指定偏移量处取出四个字节长度的数据，它记录的是消息的字节长度；</li>
     * <li>然后从内存映射文件中指定偏移量处取出指定字节长度的数据，并解码为MessageExt，并返回。</li>
     * <li>如果找不到，则返回null值。</li>
     * </ol>
     * @param commitLogOffset
     * @return
     */
    public MessageExt lookMessageByOffset(final long commitLogOffset) {
        return this.store.lookMessageByOffset(commitLogOffset);
    }

    public BrokerController getBrokerController() {
        return brokerController;
    }

    public boolean escapeMessage(MessageExtBrokerInner messageInner) {
        PutMessageResult putMessageResult = this.brokerController.getEscapeBridge().putMessage(messageInner);
        if (putMessageResult != null && putMessageResult.isOk()) {
            return true;
        } else {
            LOGGER.error("Escaping message failed, topic: {}, queueId: {}, msgId: {}",
                messageInner.getTopic(), messageInner.getQueueId(), messageInner.getMsgId());
            return false;
        }
    }
}
